//package com.atguigu;
//
//import com.ververica.cdc.connectors.mysql.source.MySqlSource;
//import com.ververica.cdc.connectors.mysql.table.StartupOptions;
//import com.ververica.cdc.debezium.DebeziumSourceFunction;
//import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.runtime.state.filesystem.FsStateBackend;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//
//public class DataStreamApi2 {
//
//    public static void main(String[] args) throws Exception {
//
//        System.setProperty("HADOOP_USER_NAME", "atguigu");
//
//        //TODO 1.获取执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//
//        //TODO 2.使用FlinkCDC创建SourceFunction
//        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
//                .hostname("hadoop102")
//                .port(3306)
//                .username("root")
//                .password("000000")
//                .databaseList("gmall-210927-flink")
//                .tableList("gmall-210927-flink.base_trademark") //注意：写表的时候一定要带上库名
//                .deserializer(new JsonDebeziumDeserializationSchema())
//                //earliest:需要再创建数据库之前就开启Binlog
//                .startupOptions(StartupOptions.initial())
//                .build();
//
//        //TODO 3.读取数据
//        DataStreamSource<String> dataStream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL");
//
//        //TODO 4.打印数据
//        dataStream.print();
//
//        //TODO 5.启动任务
//        env.execute();
//
//    }
//
//}
